package com.sdp.irrigation.server.config;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.sdp.irrigation.kafka.producer.KafkaProducer;
import com.sdp.irrigation.kafka.property.ServerProperty;
import com.sdp.irrigation.server.netty.channel.ChannelManager;
import com.sdp.irrigation.server.netty.channel.IChannelManager;
import com.sdp.irrigation.server.netty.decode.JsonMessageDecoder;
import com.sdp.irrigation.server.netty.encode.JsonMessageEncoder;
import com.sdp.irrigation.server.netty.handle.ServerHandler;
import com.sdp.irrigation.server.netty.response.IResponseHandle;
import com.sdp.irrigation.server.netty.response.JsonResponseHandle;
import com.sdp.irrigation.server.netty.server.NettyServer;
import com.sdp.irrigation.server.property.NettyProperty;
import com.sdp.irrigation.utils.CommConstants;

import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.timeout.IdleStateHandler;

@Configuration
public class NettyConfiguration {

	private static final int connectReadTimeOut = 90;
	
	private static final int maxProtocolLength = 2048;
 
    @Bean
    public NettyServer nettyServer(NettyProperty nettyProperty, ChannelInitializer<SocketChannel> channelInitializer,
    		IChannelManager channelService)
    {
    	return new NettyServer(nettyProperty, channelInitializer, channelService);
    }
    
	@Bean
	public JsonMessageDecoder jsondecoder()
	{
		JsonMessageDecoder decoder = new JsonMessageDecoder();
		return decoder;
	}
	
	@Bean
	public JsonMessageEncoder jsonencoder()
	{
		JsonMessageEncoder encoder = new JsonMessageEncoder();	
		return encoder;
	}
	
	@Bean
	public IResponseHandle jsonResponseHandle(IChannelManager channelService) {
		return new JsonResponseHandle(channelService);
	}
	
	@Bean
	public ServerHandler jsonHandler(IChannelManager channelService, Map<String, String> eqMap, IResponseHandle jsonResponseHandle, 
			KafkaProducer kafkaProducer, ServerProperty serverProperty)
	{
		return new ServerHandler(channelService, eqMap, jsonResponseHandle, kafkaProducer, serverProperty);
	}
	
	@Bean
	public Map<String, String> eqMap()
	{
		return new ConcurrentHashMap<>();
	}
	
	@Bean(name = "handleGroup")
	public EventLoopGroup handleGroup(NettyProperty nettyProperty)
	{
		return new NioEventLoopGroup(nettyProperty.getWorkerThread());
	}
	
	@Bean
	public ChannelInitializer<SocketChannel> channelInitializer(EventLoopGroup handleGroup, IChannelManager channelService, Map<String, String> eqMap,
			IResponseHandle jsonResponseHandle, KafkaProducer kafkaProducer, ServerProperty serverProperty)
	{
		return new ChannelInitializer<SocketChannel>(){

			@Override
			protected void initChannel(SocketChannel ch) throws Exception {
				ch.pipeline().addLast(new IdleStateHandler(connectReadTimeOut, 0, 0))
					.addLast(new DelimiterBasedFrameDecoder(maxProtocolLength, Unpooled.wrappedBuffer(CommConstants.JSON_DELIMITER.getBytes("UTF-8"))))
					.addLast(new JsonMessageEncoder())
					.addLast(new JsonMessageDecoder())
					.addLast(handleGroup, jsonHandler(channelService, eqMap, jsonResponseHandle, kafkaProducer, serverProperty));
//					.addLast(exceptionHandler());
			}
			
		};
	}
	
	@Bean
	public IChannelManager channelService(ServerProperty property) {
		return new ChannelManager(property.getMaxConn());
	}
}
